package net.bwie.realtime.jtp.dws.log.funcation;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import net.bwie.realtime.jtp.common.utils.DateTimeUtil;
import net.bwie.realtime.jtp.dws.log.bean.PageViewBean;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;

import java.util.HashMap;

public class PageViewBeanMapFunction extends RichMapFunction<String, PageViewBean> {

    ValueState<String> lastDateState=null;

    /**
     * 预先加载地区字段表数据，放到Map集合中
     */
    private HashMap<String, String> getAreaDic() {
        // map集合
        HashMap<String, String> map = new HashMap<>();
        // 添加数据
        map.put("110000","北京");
        map.put("120000","天津");
        map.put("140000","山西");
        map.put("150000","内蒙古");
        map.put("130000","河北");
        map.put("310000","上海");
        map.put("320000","江苏");
        map.put("330000","浙江");
        map.put("340000","安徽");
        map.put("350000","福建");
        map.put("360000","江西");
        map.put("370000","山东");
        map.put("710000","台湾");
        map.put("230000","黑龙江");
        map.put("220000","吉林");
        map.put("210000","辽宁");
        map.put("610000","陕西");
        map.put("620000","甘肃");
        map.put("630000","青海");
        map.put("640000","宁夏");
        map.put("650000","新疆");
        map.put("410000","河南");
        map.put("420000","湖北");
        map.put("430000","湖南");
        map.put("440000","广东");
        map.put("450000","广西");
        map.put("460000","海南");
        map.put("810000","香港");
        map.put("820000","澳门");
        map.put("510000","四川");
        map.put("520000","贵州");
        map.put("530000","云南");
        map.put("500000","重庆");
        map.put("540000","西藏");
        // 返回集合
        return map ;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<String> stringValueStateDescriptor = new ValueStateDescriptor<>("visit-date", String.class);
        //  设置TTL
        StateTtlConfig ttlConfig = StateTtlConfig
                .newBuilder(Time.days(1))
                .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite)
                .neverReturnExpired()
                .build();
        stringValueStateDescriptor.enableTimeToLive(ttlConfig);

        lastDateState = getRuntimeContext()
                .getState(stringValueStateDescriptor);
    }



    @Override
    public PageViewBean map(String value) throws Exception {
        JSONObject jsonkeyedStream = JSON.parseObject(value);
        JSONObject common = jsonkeyedStream.getJSONObject("common");
        JSONObject page = jsonkeyedStream.getJSONObject("page");

        long uvCount = 0L;

        Long ts = jsonkeyedStream.getLong("ts");
        String visitDate = DateTimeUtil.convertLongToString(ts, DateTimeUtil.DATE_FORMAT);
        //获取上一次的访问时间
        String lastVisitDate = lastDateState.value();
        //值为null时，表示第一次访问，uv=1
        //值不为null时，表示不是第一次访问，uv=0
        //值为null且访问时间不一致时，表示今天第一次访问，uv=1
        //值为null且访问时间一致时，表示今天不是第一次访问，uv=0
        if (null==lastVisitDate || !lastVisitDate.equals(visitDate)){
            uvCount=1L;
            lastDateState.update(visitDate);
        }



        return new PageViewBean(
                null,
                null,
                common.getString("ba"),
                common.getString("ch"),
                getAreaDic().get(common.getString("ar")),
                common.getString("is_new"),
                page.getString("last_page_id")==null  ? 1L : 0L,
                1L,
                page.getLong("during_time"),
                uvCount,
                ts
        );
    }

    @Override
    public void close() throws Exception {
        super.close();
    }
}
